/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package testing;

import cfar.actors.Node;
import cfar.actors.distributed.ActorServer;
import cfar.ff.data.Assocs.Assoc2;
import cfar.ff.data.List;
import cfar.ff.fun.F1;
import cfar.ff.fun.Seq;
import cfar.ff.interfaces.Sequence;
import cfar.utils.IO;
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.Random;
import java.lang.Math;

import cfar.actors.Message;
import cfar.ff.fun.F0;
import cfar.ff.fun.F2;

import static cfar.ff.fun.Curry.*;

/**
 *
 * @author haoyun
 */
public class Eval {

    //generate trajectories from an initial value
    public static List<Double> genTrajMap(Double iniX){
        List<Double> trajX=List.list(iniX);
        int timestep=100;
        double vn=0;
        double xn=iniX;
        double vhalf;
        double gamma=6.4114;
        double deltat=0.14;
        double delta=Math.sqrt((2*0.616*gamma)/(deltat/2));
        Random rand=new Random();
        double R=rand.nextGaussian()*delta;
        for(int i=1;i<timestep;i++){
            vhalf=vn+(deltat/2)*(-xn-gamma*vn+R);
            xn=xn+deltat*vhalf;
            R=rand.nextGaussian()*delta;
            vn=vhalf+deltat/2*(-xn-gamma*vn+R);
            trajX=trajX.snoc(xn);
        }
        return trajX;
    }

    //mark the state 1 and state 2
    public static Integer stateMap (Double x){
        int state=0;
        if(x<-1&x>-2)
            state=-1;
        else if(x>1&x<2)
            state=-2;
        else
            state=0;
        return state;
    }

    //find out whether state 1 exists before state 2
    public static Integer stateRed (Integer s1, Integer s2){
        int state=0;
        if(s1==0|s2==0)
            state=s1+s2;
        else if(s1==-2&s2==-1)
            state=-2;
        else if(s1==-1&s2==-2)
            state=1;
        else if(s1==-1&s2==-1)
            state=-1;
        else if(s1==-2&s2==-2)
            state=-2;
        else if(s1==1|s2==1)
            state=1;
        else
            System.out.println("Algorithm goes run state="+s1+", "+s2);
        return state;
    }

    //calculate the total number of trajectories in which state 1 exists before state 2
    public static Integer calProRed(Integer s1, Integer s2){
        if(s1<0)
            s1=0;
        if(s2<0)
            s2=0;
        return (s1+s2);
    }

    //build first class function for map reduce
    public static F1<Double, Integer> stateMapper = new F1<Double, Integer>() {

        public Integer f(Double a) {
            return stateMap(a);
        }
    };

    public static F2<Integer,Integer,Integer> stateReducer = new F2<Integer, Integer, Integer>() {

        public Integer f(Integer a, Integer b) {
            return stateRed(a,b);
        }
    };

    public static F1<Double, java.util.List> trajMapper =new F1<Double, java.util.List>() {

        public java.util.List f(Double x){
            List l = Eval.genTrajMap(x);
            return new java.util.LinkedList(l.getList().toCollection());
        }
    };

    public static F2<Integer, Integer, Integer> calProReducer =new F2<Integer, Integer, Integer>() {

        public Integer f(Integer s1, Integer s2){
            return Eval.calProRed(s1, s2);
        }
    };

    //the inner map reduce, for one trajectory, figure out whether this trajectory have state 1 before state 2
    //if yes, return 1, else return number 0,-1,-2
    public static F1<java.util.List<Double>, Integer> innerMR = new F1<java.util.List<Double>, Integer>() {

        public Integer f(java.util.List<Double> traj) {
//            List<Node> nodes = (List<Node>) Seq.parMap(mknode, IO.available_hosts(syncdir, suffix));
            List trajInput= (List) Seq.mkSeq(traj, List.<Double>nil());
//            List trajInput=new List();
//            for (Double o: traj){
//                trajInput=trajInput.snoc(o);
//            }
            List<Integer> trajStates = (List<Integer>) Seq.parMap(stateMapper, trajInput);

            F0 getnodes = curry(curry(get_nodes, syncdir), suffix);
            Integer state=Seq.parReduce(stateReducer, trajStates);

            return state;
        }
    };


    public static F1<Assoc2<InetAddress, Integer>, Node> mknode = new F1<Assoc2<InetAddress, Integer>, Node>() {

        public Node f(Assoc2<InetAddress, Integer> a) {
            try {
                return new Node(a.$1, a.$2);
            } catch (IOException ex) {
                Logger.getLogger(Eval.class.getName()).log(Level.SEVERE, null, ex);
            }
            return null;
        }
    };


    public static F2<String,String,Sequence<List,Node>> get_nodes = new F2<String, String, Sequence<List,Node>>() {

        public Sequence<List,Node> f(String dir, String suffix) {
            Sequence<List,Assoc2<InetAddress,Integer>> locs = IO.available_hosts(syncdir, suffix);
            List<Node> ns = List.nil();
            for(Assoc2<InetAddress,Integer> l : locs){
                try {
                    ns = ns.cons(new Node(l.$1, l.$2));
                } catch (IOException ex) {
                    Logger.getLogger(EvalSimp.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
            return ns;
        }
    };

    static String syncdir = "/tmp/cfar.sync"; // "/dscratch/cabdulwa/os/cfar.sync";
    static String suffix  = "cfar";

    public static void main(String[] args) throws IOException, InterruptedException {

 //       if(args.length == 0){
 //           for (int i = 0; i < 10; i++)
 //               new Thread(new ActorServer(syncdir, suffix)).start();
  //      }
  //      else {
            List l = List.list(1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0);
            //List l = List.list(0.0,1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0);;

            Sequence<List,Assoc2<InetAddress, Integer>> locs = IO.available_hosts(syncdir, suffix);
//            List<Node> nodes = List.nil();
//            for(Assoc2 a : locs)
//                nodes = nodes.cons(new Node((InetAddress) a.$1, (Integer) a.$2));

//            Sequence traj=Seq.distMap(nodes, trajMapper, l);
//            Sequence states=Seq.distMap(nodes, innerMR, (List)traj);
//            F0 getnodes = curry(curry(get_nodes, syncdir), suffix);
//            Integer factor=Seq.distReduce(getnodes, calProReducer, states);
            Sequence traj = Seq.parMap(trajMapper, l);
            Sequence states = Seq.parMap(innerMR, traj);
            int factor = Seq.parReduce(calProReducer, states);


 //           for(Node n : nodes)
 //               n.tell(Message.stop);

            double prob=(double)factor/traj.length();
            System.out.println("The probability that state a exists before state b is "+prob);
            System.out.println("Done.");
            System.exit(0);
//        }
    }


}
